package logger

import (
	"errors"

	"github.com/segmentio/kafka-go"
)

// Connection 连接
type Connection struct {
	writer  *kafka.Writer
	servers []string
	topic   string
}

// NewConnection 连接kafka
func NewConnection(servers []string, topic string) (*Connection, error) {

	if servers == nil || len(servers) == 0 || topic == "" {
		return nil, errors.New("params error")
	}
	writer := kafka.NewWriter(kafka.WriterConfig{
		Brokers:  servers,
		Topic:    topic,
		Balancer: &kafka.LeastBytes{},
	})

	return &Connection{
		writer:  writer,
		servers: servers,
		topic:   topic,
	}, nil
}

// Close 关闭连接
func (conn *Connection) Close() error {
	return conn.writer.Close()
}

// Reload 重新连接
func (conn *Connection) Reload() error {
	conn.writer.Close()
	writer := kafka.NewWriter(kafka.WriterConfig{
		Brokers:  conn.servers,
		Topic:    conn.topic,
		Balancer: &kafka.LeastBytes{},
	})
	conn.writer = writer
	return nil
}
